package com.seari.kafka;

import java.io.UnsupportedEncodingException;
import java.util.Date;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.seari.bean.KafkaMessage;

@Component
public class KafkaSender
{
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;
	
	ObjectMapper mapper = new ObjectMapper();
	
	protected static Logger logger = LoggerFactory.getLogger(KafkaSender.class);
	
	public void send()
	{
		KafkaMessage message = new KafkaMessage();
		message.setUuid(UUID.randomUUID().toString());
		message.setMessage("test message");
		message.setTimeStamp(new Date());
		byte[] str;
		String result = "";
		try
		{
			str = mapper.writeValueAsBytes(message);
			result = new String(str,"utf-8");
			logger.info("kafka sender: message{}",result);
			kafkaTemplate.send("test2",result);
		} catch (JsonProcessingException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (UnsupportedEncodingException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally {
			str = null;
			result = null;
		}
	}
	

	public void send(Object message,String topic) 
	{
		byte[] str;
		String result = "";
		try
		{
			str = mapper.writeValueAsBytes(message);
			result = new String(str,"utf-8");
			logger.info("kafka sender: message{}",result);
			kafkaTemplate.send(topic,result);
		} catch (JsonProcessingException e)
		{
			e.printStackTrace();
		} catch (UnsupportedEncodingException e)
		{
			e.printStackTrace();
		}finally {
			str = null;
			result = null;
		}
	}
}
